package com.gallop.mq.core.dispatcher;

import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.io.Serializable;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * MQ rpc模式响应回调
 */
@Slf4j
public class DispatchRpcCallable implements Serializable {

    private DispatchRpcCallable() {
        this.finished = new AtomicBoolean();
    }

    private final AtomicBoolean finished;
    /**
     * rpc模式下，响应消息需要发送至的topic
     */
    private String replyTo;
    /**
     * rpc模式下，请求关联ID
     */
    private String correlationId;
    /**
     * rpc模式下响应回调
     * 参数： 响应topic和响应消息
     */
    private DispatchContext.RpcResponse rpcResponse;

    public void reply(Object responseMessage) {
        if (!StringUtils.hasLength(replyTo)) {
            return;
        }
        if (!finished.compareAndSet(false, true)) {
            return;
        }
        rpcResponse.accept(correlationId, replyTo, responseMessage);
    }

    public static DispatchRpcCallableBuilder builder() {
        return new DispatchRpcCallableBuilder();
    }

    /**
     * 创建一个永远不会被执行的响应回调
     */
    public static DispatchRpcCallable nop() {
        DispatchRpcCallable ret = new DispatchRpcCallable();
        ret.finished.set(true);
        ret.rpcResponse = (correlationId, replyTo, resp) -> log.warn("MQ nop-stub rpc callback executed. correlation={} reply={},resp={}", correlationId, replyTo, resp);
        return ret;
    }

    @Override
    public String toString() {
        return "DispatchRpcCallable{" +
                "finished=" + finished +
                ", replyTo='" + replyTo + '\'' +
                ", correlationId='" + correlationId + '\'' +
                ", rpcResponse=" + rpcResponse +
                '}';
    }

    @Getter
    public static class DispatchRpcCallableBuilder {
        private String replyTo;

        private String correlationId;
        private DispatchContext.RpcResponse rpcResponse;

        public DispatchRpcCallableBuilder replyTo(String replyTo) {
            this.replyTo = replyTo;
            return this;
        }

        public DispatchRpcCallableBuilder correlationId(String correlationId) {
            this.correlationId = correlationId;
            return this;
        }

        public DispatchRpcCallableBuilder rpcResponse(DispatchContext.RpcResponse rpcResponse) {
            this.rpcResponse = rpcResponse;
            return this;
        }

        public DispatchRpcCallable build() {
            DispatchRpcCallable ret = new DispatchRpcCallable();
            ret.replyTo = this.replyTo;
            ret.correlationId = this.correlationId;
            ret.rpcResponse = this.rpcResponse;
            return ret;
        }
    }
}